package org.fly.jingwei.service.impl;

import org.fly.jingwei.dao.SlaveDao;
import org.fly.jingwei.dao.TaskGroupDao;
import org.fly.jingwei.entity.JobEntity;
import org.fly.jingwei.entity.SlaveEntity;
import org.fly.jingwei.entity.TaskControlEntity;
import org.fly.jingwei.entity.TransformationEntity;
import org.fly.jingwei.ext.JobExecutor;
import org.fly.jingwei.ext.TransExecutor;
import org.fly.jingwei.service.ControlService;
import org.fly.jingwei.util.KettleEncrypt;
import org.fly.jingwei.util.task.CarteClient;
import org.pentaho.di.cluster.SlaveServer;
import org.pentaho.di.job.JobExecutionConfiguration;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransExecutionConfiguration;
import org.pentaho.di.trans.step.StepStatus;
import org.pentaho.di.www.SlaveServerJobStatus;
import org.pentaho.di.www.SlaveServerStatus;
import org.pentaho.di.www.SlaveServerTransStatus;
import org.pentaho.di.www.StopTransServlet;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.List;

/**
 * @author leo.jie (weixiao.me@aliyun.com)
 * @version 1.0
 * @organization bigdata
 * @website https://www.jielongping.com
 * @date 2020/5/30 10:58 上午
 * @since 1.0
 */
@Service
public class ControlServiceImpl  implements ControlService {
    //extends StopTransServlet
    private final SlaveDao slaveDao;
    private final TaskGroupDao taskGroupDao;

    public ControlServiceImpl(SlaveDao slaveDao, TaskGroupDao taskGroupDao) {
        this.slaveDao = slaveDao;
        this.taskGroupDao = taskGroupDao;
    }

    @Override
    public List<TaskControlEntity> getAllRunningJob(String userGroupName) throws Exception {
        //获取所有运行中的作业
        List<JobEntity> jobItems = taskGroupDao.getAllJob(userGroupName);
        List<TaskControlEntity> items = new ArrayList<>();
        //获取本地执行的作业
        Hashtable<String, JobExecutor> table = JobExecutor.getExecutors();
        Enumeration<String> keys = table.keys();
        while (keys.hasMoreElements()) {
            String jobExecutorId = keys.nextElement();
            JobExecutor jobExecutor = table.get(jobExecutorId);
            JobExecutionConfiguration executionConfiguration = jobExecutor.getExecutionConfiguration();
            //作业未完成 而且是本地运行
            if (!jobExecutor.isFinished() && executionConfiguration.isExecutingLocally() && !jobExecutor.isClickStop()) {
                //判断当前作业是否是该用户的可见作业
                boolean isSee = false;
                for (JobEntity thisJob : jobItems) {
                    if (thisJob.getName().equals(jobExecutor.getJobMeta().getName())) {
                        isSee = true;
                        break;
                    }
                }
                if (!isSee) {
                    continue;
                }
                TaskControlEntity taskControl = new TaskControlEntity();
                taskControl.setName(jobExecutor.getJobMeta().getName());
                taskControl.setId(jobExecutor.getExecutionId());
                taskControl.setType("作业");
                taskControl.setIsStart("");
                taskControl.setCarteObjectId(jobExecutor.getCarteObjectId());
                taskControl.setHostName("本地执行");
                items.add(taskControl);
            }
        }
        //从carte服务获取远程执行的作业  从所有节点上获取正在运行的作业
        List<SlaveEntity> slaves = slaveDao.getAllSlave("");
        for (SlaveEntity slave : slaves) {
            slave.setPassword(KettleEncrypt.decryptPassword(slave.getPassword()));
            CarteClient carte = new CarteClient(slave);
            String status = carte.getStatusOrNull();
            if (status == null) {
                continue;
            }
            SlaveServerStatus serverStatus = SlaveServerStatus.fromXML(status);
            for (SlaveServerJobStatus jobStatus : serverStatus.getJobStatusList()) {
                if (jobStatus.isRunning()) {
                    TaskControlEntity item = new TaskControlEntity();
                    String jobStatusxml = carte.getJobStatus(jobStatus.getJobName(), jobStatus.getId());
                    SlaveServerJobStatus realJobStatus = SlaveServerJobStatus.fromXML(jobStatusxml);
                    //判断当前作业是否是该用户的可见作业
                    boolean isSee = false;
                    for (JobEntity thisJob : jobItems) {
                        if (thisJob.getName().equals(realJobStatus.getJobName())) {
                            isSee = true;
                            break;
                        }
                    }
                    if (!isSee) {
                        continue;
                    }
                    //根据前台展示需求来封装 暂时只展示作业名以及所运行节点
                    item.setHostName(slave.getHostName());
                    item.setId(realJobStatus.getId());
                    item.setName(jobStatus.getJobName());
                    item.setType("作业");
                    item.setIsStart("");
                    items.add(item);
                }
            }
        }
        return items;
    }

    @Override
    public List<TaskControlEntity> getAllRunningTrans(String userGroupName) throws Exception {
        //获取所有运行中的转换
        //获取该用户组下所有可见的转换
        List<TransformationEntity> transItems = taskGroupDao.getAllTrans(userGroupName);

        List<TaskControlEntity> items = new ArrayList<>();
        //获取内存中正在运行的转换   本地执行的转换
        Hashtable<String, TransExecutor> table = TransExecutor.getExecutors();
        Enumeration<String> keys = table.keys();
        while (keys.hasMoreElements()) {
            TransExecutor transExecutor = table.get(keys.nextElement());
            TransExecutionConfiguration executionConfiguration = transExecutor.getExecutionConfiguration();
            SlaveServer slaveServer = executionConfiguration.getRemoteServer();
            Trans trans = transExecutor.getTrans();
            //转换未完成并且是在本地执行
            if (!transExecutor.isFinished() && executionConfiguration.isExecutingLocally() && !transExecutor.isClickStop()) {
                //判断当前转换是否是该用户的可见转换
                boolean isSee = false;
                for (TransformationEntity transformation : transItems) {
                    if (transformation.getName().equals(transExecutor.getTransMeta().getName())) {
                        isSee = true;
                        break;
                    }
                }
                if (!isSee) {
                    continue;
                }
                //本地执行
                TaskControlEntity taskControl = new TaskControlEntity();
                taskControl.setName(transExecutor.getTransMeta().getName());
                taskControl.setId(transExecutor.getExecutionId());
                taskControl.setType("转换");
                taskControl.setCarteObjectId(transExecutor.getCarteObjectId());
                taskControl.setHostName("本地执行");
                if (trans.isPaused()) {
                    taskControl.setIsStart("暂停中");
                } else {
                    taskControl.setIsStart("正在运行");
                }
                items.add(taskControl);
            }
        }
        //获取远程执行的转换 从carte服务上获取
        List<SlaveEntity> slaves = slaveDao.getAllSlave("");
        for (SlaveEntity slave : slaves) {
            //对数据库里取出的密码进行转码
            slave.setPassword(KettleEncrypt.decryptPassword(slave.getPassword()));
            CarteClient carte = new CarteClient(slave);
            String status = carte.getStatusOrNull();
            if (status == null) {
                continue;
            }
            SlaveServerStatus serverStatus = SlaveServerStatus.fromXML(status);
            for (SlaveServerTransStatus transStatus : serverStatus.getTransStatusList()) {
                //转换是运行或者暂停状态
                if (transStatus.isRunning() || transStatus.isPaused()) {
                    TaskControlEntity item = new TaskControlEntity();
                    String transStatusXml = carte.getTransStatus(transStatus.getTransName(), transStatus.getId());
                    SlaveServerTransStatus realTransStatus = SlaveServerTransStatus.fromXML(transStatusXml);
                    //判断当前转换是否是该用户的可见转换
                    boolean isSee = false;
                    for (TransformationEntity transformation : transItems) {
                        if (transformation.getName().equals(realTransStatus.getTransName())) {
                            isSee = true;
                            break;
                        }
                    }
                    if (!isSee) {
                        continue;
                    }
                    item.setType("转换");
                    item.setHostName(slave.getHostName());
                    item.setName(realTransStatus.getTransName());
                    item.setId(transStatus.getId());
                    if (transStatus.isRunning()) {
                        item.setIsStart("正在运行");
                    } else if (transStatus.isPaused()) {
                        item.setIsStart("暂停中");
                    }
                    items.add(item);
                }
            }
        }
        return items;
    }

    @Override
    public String getLogDetailForJob(String id, String hostName) throws Exception {
        //获取作业日志
        String log = "";
        if ("本地执行".equals(hostName)) {
            Hashtable<String, JobExecutor> table = JobExecutor.getExecutors();
            Enumeration<String> keys = table.keys();
            while (keys.hasMoreElements()) {
                String jobExecutorId = keys.nextElement();
                JobExecutor jobExecutor = table.get(jobExecutorId);
                if (jobExecutor.getExecutionId().equals(id)) {
                    log = jobExecutor.getExecutionLog();
                    log = log.replaceAll("\n", "<br/>");
                }
            }
        } else {
            SlaveEntity slave = slaveDao.getSlaveByHostName(hostName);
            slave.setPassword(KettleEncrypt.decryptPassword(slave.getPassword()));
            CarteClient cc = new CarteClient(slave);
            log = cc.getJobLogText(id);
            log = log.replaceAll("\n", "<br/>");
        }
        return log;
    }

    @Override
    public String getLogDetailForTrans(String id, String hostName) throws Exception {
        //获取转换日志
        String result = "";
        if ("本地执行".equals(hostName)) {
            Hashtable<String, TransExecutor> table = TransExecutor.getExecutors();
            Enumeration<String> keys = table.keys();
            while (keys.hasMoreElements()) {
                String transExecutorId = keys.nextElement();
                TransExecutor transExecutor = table.get(transExecutorId);
                if (id.equals(transExecutor.getExecutionId())) {
                    result = transExecutor.getExecutionLog();
                    result = result.replaceAll("\n", "<br/>");
                }
            }
        } else {
            SlaveEntity slave = slaveDao.getSlaveByHostName(hostName);
            slave.setPassword(KettleEncrypt.decryptPassword(slave.getPassword()));
            CarteClient cc = new CarteClient(slave);
            result = cc.getTransLogText(id);
            result = result.replaceAll("\n", "<br/>");
        }
        return result;
    }

    @Override
    public List<StepStatus> getTransDetail(String id, String hostName) throws Exception {
        List<StepStatus> result = null;
        if ("本地执行".equals(hostName)) {
            Hashtable<String, TransExecutor> table = TransExecutor.getExecutors();
            Enumeration<String> keys = table.keys();
            while (keys.hasMoreElements()) {
                String transExecutorId = keys.nextElement();
                TransExecutor transExecutor = table.get(transExecutorId);
                if (transExecutor.getExecutionId().equals(id)) {
                    result = transExecutor.getTransStepStatus();
                }
            }
        } else {
            SlaveEntity slave = slaveDao.getSlaveByHostName(hostName);
            slave.setPassword(KettleEncrypt.decryptPassword(slave.getPassword()));
            CarteClient carte = new CarteClient(slave);
            //获取当前节点下所有正在运行的转换
            String status = carte.getStatusOrNull();
            SlaveServerStatus serverStatus = SlaveServerStatus.fromXML(status);
            //遍历这些转换,判断和所需要查询的ID是否相同
            for (SlaveServerTransStatus transStatus : serverStatus.getTransStatusList()) {
                //如果转换没有停止
                if (transStatus.isRunning() || transStatus.isPaused()) {
                    String transStatusXml = carte.getTransStatus(transStatus.getTransName(), transStatus.getId());
                    SlaveServerTransStatus realTransStatus = SlaveServerTransStatus.fromXML(transStatusXml);
                    //如果相同则获取当前正在运行的转换详情
                    if (realTransStatus.getId().equals(id.trim())) {
                        result = realTransStatus.getStepStatusList();
                    }
                }
            }
        }
        return result;
    }

    @Override
    public void stopTrans(String hostName, String id) throws Exception {
        if("本地执行".equals(hostName)){
            Hashtable<String,TransExecutor> table= TransExecutor.getExecutors();
            Enumeration<String> keys=table.keys();
            while (keys.hasMoreElements()){
                String transExecutorId = keys.nextElement();
                TransExecutor transExecutor = table.get(transExecutorId);
                if(transExecutor.getExecutionId().equals(id) && !transExecutor.isFinished() && !transExecutor.isClickStop()){
                    transExecutor.stop();
                    transExecutor.setIsClickStop(true);
                    return;
                }
            }
        }else{
            SlaveEntity slave=slaveDao.getSlaveByHostName(hostName);
            slave.setPassword(KettleEncrypt.decryptPassword(slave.getPassword()));
            CarteClient cc=new CarteClient(slave);
            cc.stopTrans(id);
        }
    }

    @Override
    public void stopJob(String hostName, String id) throws Exception {
        if("本地执行".equals(hostName)){
            Hashtable<String,JobExecutor> table= JobExecutor.getExecutors();
            Enumeration<String> keys=table.keys();
            while (keys.hasMoreElements()){
                String jobExecutorId = keys.nextElement();
                JobExecutor jobExecutor = table.get(jobExecutorId);
                if(jobExecutor.getExecutionId().equals(id)){
                    jobExecutor.stop();
                    jobExecutor.setIsClickStop(true);
                }
            }
        }else{
            SlaveEntity slave=slaveDao.getSlaveByHostName(hostName);
            slave.setPassword(KettleEncrypt.decryptPassword(slave.getPassword()));
            CarteClient cc=new CarteClient(slave);
            cc.stopJob(id);
        }
    }

    @Override
    public void pauseOrStartTrans(String[] id, String[] hostName) throws Exception {
        //暂停/开始转换
        for (int i = 0; i < id.length; i++) {
            if ("本地执行".equals(hostName[i])) {
                Hashtable<String, TransExecutor> table = TransExecutor.getExecutors();
                Enumeration<String> keys = table.keys();
                while (keys.hasMoreElements()) {
                    String transExecutorId = keys.nextElement();
                    TransExecutor transExecutor = table.get(transExecutorId);
                    if (id[i].equals(transExecutor.getExecutionId())) {
                        Trans trans = transExecutor.getTrans();
                        if (!trans.isPaused()) {
                            trans.pauseRunning();
                        } else {
                            trans.resumeRunning();
                        }
                    }
                }
            } else {
                SlaveEntity slave = slaveDao.getSlaveByHostName(hostName[i]);
                slave.setPassword(KettleEncrypt.decryptPassword(slave.getPassword()));
                CarteClient cc = new CarteClient(slave);
                cc.pauseTrans(id[i]);
            }
        }
    }

    public String transToStatusDesc(SlaveServerTransStatus realTransStatus) throws Exception {
        String running_status = null;
        if (realTransStatus.isRunning()) {
            running_status = "运行中";
        } else if (realTransStatus.isPaused()) {
            running_status = "暂停";
        } else {
            running_status = realTransStatus.getStatusDescription();
        }
        return running_status;
    }
}
